package com.jse.jdbc;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;

import com.jse.Io;
import com.jse.Log;

/**
     * 首先将数据库数据转换成java bean,然后再将java bean放入Spliterator中.
     *
     * @param <T>
     *         Spliterator中包含的元素,也就是转换得到的java bean的类型.
     * @author wuda
     */
public class RsSpliterator<T> implements Spliterator<T> {

	private static Log logger = Log.getLogger(RsSpliterator.class);
        /**
         * sql语句.
         */
        private String sql;
        /**
         * {@link ResultSet}转java bean的转换器.
         */
        private Function<ResultSet, T> rowMapper;
        /**
         * 分隔单位,每达到当前指定的分隔大小时,进行一次分隔.
         */
        private int splitSize = 128;

        /**
         * 数据库连接对象.
         */
        private Connection connection;
        private Object[] args;
        /**
         * statement.
         */
        private PreparedStatement statement;
        /**
         * result set.
         */
        private ResultSet resultSet;

        /**
         * 生成spliterator.使用{@link ResultSet}作为数据来源,并且使用<code>rowMapper</code>可以将数据转换成java
         * bean.
         *
         * @param dataSource
         *         数据源
         * @param sql
         *         sql语句
         * @param rowMapper
         *         将数据库数据映射为实体
         */
        public RsSpliterator(String sql,Function<ResultSet, T> rowMapper,Object...args) {
            this.sql = sql;
            this.rowMapper = rowMapper;
            this.args=args;
        }

        @Override
        public boolean tryAdvance(Consumer<? super T> action) {
            // 把握一切机会关闭数据库资源
            Jdbc.close(connection, statement, resultSet);
            // 全部都分出去了,这里不会有待处理元素
            return false;
        }

        @Override
        public Spliterator<T> trySplit() {
            if (resultSet == null && acquireChanceToExecuteQuery()) {
                try {
                    connection = Jdbc.connection();
                    connection.setReadOnly(true);
                    statement = connection.prepareStatement(sql,ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
                    if(args!=null&&args.length>0) {
                    	for (int i = 0; i < args.length; i++) {
                    		statement.setObject(i+1,args[i]);
						}
                    }
                    statement.setFetchSize(Integer.MIN_VALUE);
                    resultSet = statement.executeQuery();
//                    logger.error("resultSet={}", resultSet);
                } catch (SQLException e) {
                    logger.error(e.getMessage(), e);
                    Jdbc.close(connection, statement, resultSet);
                    return null;
                }
            }
            try {
                if (!resultSet.next()) {
                	Jdbc.close(connection, statement, resultSet);
                    return null;
                }
            } catch (Exception e) {
            	Jdbc.close(connection, statement, resultSet);
                logger.warn("resultSet.next()方法调用时异常,{}", e.getMessage(), e);
                return null;
            }
            int count = 0;
            Object[] array = new Object[splitSize];
            try {
                do {
                    array[count] = rowMapper.apply(resultSet);
                    count++;
                } while (count < splitSize && resultSet.next());
            } catch (SQLException e) {
                logger.warn("resultSet.next()方法调用时异常,{}", e.getMessage(), e);
                if (count == 0) {
                	Jdbc.close(connection, statement, resultSet);
                    return null;
                }
            }
            return Spliterators.spliterator(array, count);
        }

        @Override
        public long estimateSize() {
            return Long.MAX_VALUE;
        }

        @Override
        public int characteristics() {
            return 4096;
        }

        /**
         * sql语句是否已经执行.
         */
        private AtomicBoolean queryExecuted = new AtomicBoolean(false);

        /**
         * 获取执行sql语句的机会.
         *
         * @return true-可已执行,false-不能执行
         */
        private boolean acquireChanceToExecuteQuery() {
            while (!queryExecuted.get()) {
                if (queryExecuted.compareAndSet(false, true)) {
                    return true;
                }
            }
            return false;
        }

        /**
         * 关闭数据库连接的任务.
         *
         * @return task
         */
        public Runnable close() {
            return () -> Jdbc.close(connection, statement, resultSet);
        }
    }